package dowork;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import pojo.LogBean;
import pojo.ResultBean;

import java.util.ArrayList;
import java.util.Map;
import java.util.Properties;

/**
 * 疑问记录：
 * 1，liststate不会被撑爆
 *
 */
public class RuleEngineV1pre {


    /**
     * 事件流
     * {"account":"005435",
     * "appId":"cn.doitedu.yinew",
     * "appVersion":"2.5",
     * "carrier":"中国移动",
     * "deviceId":"005435",
     * "deviceType":"mi6",
     * "ip":"10.102.36.88",
     * "latitude":33.202642361456356,
     * "longitude":153.15726700124804,
     * "netType":"5G",
     * "osName":"android",
     * "osVersion":"7.5",
     * "eventId":"W",
     * "properties":{
     * "p3":"v2",
     * "p5":"v8"
     * },
     * "releaseChannel":"小米应用市场","resolution":"2048*1024"
     * }
     * <p>
     * <p>
     * <p>
     * //画像的值
     * f
     * p1         p2      p3     p4    p5
     * rowkey        v091       v9      v8     v980   v377
     *
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.enableCheckpointing(1000 * 60 * 2);
//        env.setStateBackend("")/

        //创建kafka数据流
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "master3:9092,master4:9092,master5:9092");
        properties.setProperty("group.id", "moudletest");
        //如果没有记录便偏移量，就从最开始的地方进行消费，如果记录过，就接着读。
        properties.setProperty("auto.offset.reset", "earliest");
        properties.setProperty("enable.auto.commiit", "true");
        DataStream<String> kafkastream = env.addSource(new FlinkKafkaConsumer<String>("demo_kafka_topic_1", new SimpleStringSchema(), properties));
//        kafkastream.print();

        //将过来的数据封装成一个bean
        SingleOutputStreamOperator<LogBean> logbeanstream = kafkastream.map(new MapFunction<String, LogBean>() {
            @Override
            public LogBean map(String jsonline) throws Exception {
                return JSON.parseObject(jsonline, LogBean.class);
            }
        });

        //按照deviceid对用户进行keyby
        KeyedStream<LogBean, String> logBeanKeyedStream = logbeanstream.keyBy(new KeySelector<LogBean, String>() {
            @Override
            public String getKey(LogBean logBean) throws Exception {
                return logBean.getDeviceId();
            }
        });


        //主逻辑
        logBeanKeyedStream.process(new KeyedProcessFunction<String, LogBean, ResultBean>() {
            private transient ListState<LogBean> listState;
            org.apache.hadoop.conf.Configuration conf;
            Connection conn;
            Table htable;

            @Override
            public void open(Configuration parameters) throws Exception {
                conf = new org.apache.hadoop.conf.Configuration();
                conf.set("hbase.zookeeper.quorum", "master3:2181,master4:2181,master5:2181");
                conn = ConnectionFactory.createConnection(conf);
                htable = conn.getTable(TableName.valueOf("yiniu"));

//                listState=new ListState<LogBean>(new ListStateDescriptor<LogBean>("logbean_liststate", TypeInformation.of(new TypeHint<LogBean>()){})) ;
            }

            @Override
            public void processElement(LogBean logBean, Context ctx, Collector<ResultBean> out) throws Exception {
                //先看这一条数据是否满足触发条件  eventId=W
                if ("w".equals(logBean.getEventId())) {
                    //满足，去画像库里面根据deviceid查询画像属性 需要同时满足多个属性

                    //注意每次查询，都会有一个get对象
                    Get get = new Get(Bytes.toBytes(logBean.getEventId()));
                    get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("p1"));
                    get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("p2"));
                    get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("p3"));
                    Result result = htable.get(get);

                    //从结果中取出数据
                    String v1 = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes("p1")));
                    String v2 = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes("p2")));
                    String v3 = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes("p3")));

                    if ("v400".equals(v1) && "v600".equals(v2) && "v800".equals(v3)) {
                        //查询用户的行为属性条件,那么就需要利用状态  A[p1=v3，p2=v6]=8,C[p9=v8,p6=v55]=9
                        Iterable<LogBean> logBeans = listState.get();
                        int a_i = 0;
                        int c_i = 0;

                        //计算历史以来的满足我们指定事件要求的次数
                        for (LogBean bean : logBeans) {
                            //如果是A事件
                            if ("A".equals(bean.getEventId())) {
                                Map<String, String> properties1 = bean.getProperties();
                                String p1_v = properties1.get("p1");
                                String p2_v = properties1.get("p2");
                                if ("v3".equals(p1_v) && "v6".equals(p2_v)) a_i++;

                            }
                            //如果是C事件
                            if ("C".equals(bean.getEventId())) {
                                Map<String, String> properties1 = bean.getProperties();
                                String p9_v = properties1.get("p9");
                                String p6_v = properties1.get("p6");
                                if ("v8".equals(p9_v) && "v55".equals(p6_v)) c_i++;
                            }
                        }

                        //当行为次数条件满足，就看行为次序
                        if (a_i >= 3 && c_i >= 6) {
                            System.out.println("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa");
                            //查询用户的行为序列数据  R[p1=v4]---->W(p2=v3)-->F
                            //思路：先找R,并且找到对应属性和值，可以，就往下走,以此类推
                            //      W  W W R  W   W   R  R     F
                            ArrayList<LogBean> logBeans1 = new ArrayList<>();
                            for (LogBean bean : logBeans) {
                                logBeans1.add(bean);
                            }

                            int r_index = -1;
                            for (int i = 0; i < logBeans1.size(); i++) {
                                LogBean logBean1 = logBeans1.get(i);
                                if ("R".equals(logBean1.getEventId())) {
                                    Map<String, String> properties1 = logBean1.getProperties();
                                    if ("v4".equals(properties1.get("p1"))) {
                                        r_index = i;  //记录下来
                                        break;  //如果找不到，就不会改变r_index
                                    }
                                }
                            }

                            if (r_index != -1 && r_index + 1 < logBeans1.size()) {
                                System.out.println("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");

                                int w_index = -1;
                                for (int i = 0; i < logBeans1.size(); i++) {
                                    LogBean logBean1 = logBeans1.get(i);
                                    if ("W".equals(logBean1.getEventId())) {
                                        Map<String, String> properties1 = logBean1.getProperties();
                                        if ("v3".equals(properties1.get("p2"))) {
                                            w_index = i;  //记录下来
                                            break;  //如果找不到，就不会改变r_index
                                        }
                                    }
                                }


                                if (w_index != -1 && r_index + 1 < logBeans1.size()) {
                                    int f_index = -1;
                                    for (int i = 0; i < logBeans1.size(); i++) {
                                        LogBean logBean1 = logBeans1.get(i);
                                        if ("W".equals(logBean1.getEventId())) {
                                            Map<String, String> properties1 = logBean1.getProperties();
                                            if ("v3".equals(properties1.get("p2"))) {

                                                f_index = i;  //记录下来
                                                break;  //如果找不到，就不会改变r_index
                                            }
                                        }
                                    }
                                    if (f_index != -1) {
                                        ResultBean resultBean = new ResultBean();
                                        resultBean.setDeviceid(logBean.getDeviceId());
                                        resultBean.setTimeStamp(logBean.getTimeStamp());
                                        resultBean.setRuleid("888");
                                        out.collect(resultBean);
                                    }
                                }




                            }
                        }
                    }
                }
            }
        });
        env.execute("myproject");
    }
}
